package com.cnebula.dataprocess.sqoop.job.comsumer;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.rocketmq.spring.starter.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.starter.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.cnebula.dataprocess.common.JobStatus;
import com.cnebula.dataprocess.common.sqoop.DataSourceNode;
import com.cnebula.dataprocess.common.sqoop.SqoopJobResult;
import com.cnebula.dataprocess.sqoop.job.LinkConstants;
import com.cnebula.dataprocess.sqoop.job.config.SqoopConfig;
import com.cnebula.dataprocess.sqoop.job.model.FromSqoopJobConstants;
import com.cnebula.dataprocess.sqoop.job.model.LinkConfig;
import com.cnebula.dataprocess.sqoop.job.model.LspMLink;
import com.cnebula.dataprocess.sqoop.job.model.ToSqoopJobConstants;
import com.cnebula.dataprocess.sqoop.job.producer.SqoopJobProducer;
import com.cnebula.dataprocess.sqoop.job.service.SqoopService;
import com.google.gson.Gson;

@Service
@RocketMQMessageListener(topic = "datanode-topic", consumerGroup = "datanode-consumer")
public class DataSourceNodeConsumer implements RocketMQListener<DataSourceNode> {

	private static final Logger log = LoggerFactory.getLogger(DataSourceNodeConsumer.class);

	private ExecutorService executorService = Executors.newFixedThreadPool(10);

	private Gson gson = new Gson();

	@Autowired
	SqoopConfig sqoopConf;

	@Autowired
	SqoopService sqoopService;

	@Autowired
	SqoopJobProducer producer;

	@Override
	public void onMessage(DataSourceNode node) {

		log.info("============================receive sqoop job============================================");
		log.info(gson.toJson(node));
		log.info("=========================================================================================");

		String dsType = node.getDbType();
		LinkConfig fromLinkConf = null;

		if (dsType.equalsIgnoreCase(LinkConstants.TYPE_MYSQL)) {
			fromLinkConf = getMysqlLinkConfig(node);
		}

		if (fromLinkConf == null) {
			SqoopJobResult body = new SqoopJobResult();
			body.setTaskId(node.getTaskId());
			body.setJobId(node.getJobId());
			body.setStatus(JobStatus.FAIL);
			body.setMessage("not support " + dsType + " type connection");

			producer.sendMsg(body);
			return;

		}
		final LinkConfig fromLinkConfig = fromLinkConf;

		LinkConfig toLinkConfig = new LinkConfig();
		toLinkConfig.setUrl(sqoopConf.getHadoopNamenode());
		toLinkConfig.setType(LinkConstants.TYPE_HDFS);
		toLinkConfig.setHadoopConfDir(sqoopConf.getHadoopConfDir());

		Map<String, String> fromJobConfig = new HashMap<String, String>();
		fromJobConfig.put(FromSqoopJobConstants.JDBC_SQL, node.getSql());
		fromJobConfig.put(FromSqoopJobConstants.JDBC_PARTITIONCOLUMN, node.getPartition());

		Map<String, String> toJobConfig = new HashMap<String, String>();
		toJobConfig.put(ToSqoopJobConstants.HDFS_OUTPUTDIRECTORY, sqoopConf.getJobDir());

		executorService.execute(new Runnable() {

			@Override
			public void run() {
				try {
					LspMLink fromLink = sqoopService.createMLink(fromLinkConfig);
					LspMLink toLink = sqoopService.createMLink(toLinkConfig);
					sqoopService.submitJob(node.getTaskId(), node.getJobId(), fromLink, toLink, fromJobConfig,
							toJobConfig);
				} catch (Throwable e) {
					SqoopJobResult body = new SqoopJobResult();
					body.setTaskId(node.getTaskId());
					body.setJobId(node.getJobId());
					body.setStatus(JobStatus.FAIL);
					body.setMessage(e.getMessage());

					producer.sendMsg(body);
				}
			}
		});
	}

	private LinkConfig getMysqlLinkConfig(DataSourceNode node) {

		String host = node.getHost();
		int port = node.getPort();
		String encoding = node.getCharset();
		String username = node.getUsername();
		String password = node.getPassword();
		String databaseName = node.getDatabase();

		StringBuilder url = new StringBuilder("jdbc:mysql://");
		url.append(host).append(":").append(port);
		url.append("/");
		url.append(databaseName);

		if (encoding != null && !encoding.isEmpty()) {
			url.append("?");
			url.append("useUnicode=true");
			url.append("&");
			url.append("characterEncoding=" + encoding);
		}

		LinkConfig linkConfig = new LinkConfig();
		linkConfig.setUrl(url.toString());
		linkConfig.setDriver("com.mysql.jdbc.Driver");
		linkConfig.setType(LinkConstants.TYPE_MYSQL);
		linkConfig.setUserName(username);
		linkConfig.setPassword(password);

		return linkConfig;
	}

}
